Agent Workflow（代理人工作流） 是將大型語言模型（LLM）從「單次對談機器人」升級為「自動化複雜任務系統」的藍圖與骨幹。

如果說 Agent 是具備思考與工具調用能力的個體，那麼 Workflow 就是定義這些 Agent 如何步驟化執行、控制條件分支、處理異常，以及彼此協同合作的架構與流程圖。

一、 為什麼需要 Agent Workflow？（Workflow vs. Autonomous Agent）

在 Agent 的發展歷程中，業界逐漸劃分為兩種設計哲學：

💡 Andrew Ng（吳恩達）的核心觀點：
在許多實務應用中，使用較小/較便宜的模型搭配精心設計的 Agent Workflow，其表現與可靠度往往能大幅超越直接給予強大模型（如 GPT-4o）單一提示詞（Direct Prompting）。

二、 五大核心 Agent Workflow 模式（Design Patterns）

Anthropic 與 OpenAI 在其 Agent 指南中，將常見的 Agent Workflow 歸納為以下五種基本圖拓撲（Graph Topologies）：

1. Chain (鏈式)              2. Routing (路由/分流)        3. Parallelization (平行處理)
   [Step A]                    ┌──► [Agent A]                ┌──► [Worker A] ──┐
      │          [Router] ─────┼──► [Agent B]    [Input] ────┼──► [Worker B] ──┼──► [Aggregator]
      ▼                        └──► [Agent C]                └──► [Worker C] ──┘
   [Step B]

4. Orchestrator-Workers (主從協調)            5. Evaluator-Optimizer (評估修復迴圈)
           [Orchestrator]                             ┌───────────────────────┐
          ╱      │      ╲                             ▼                       │
   [Worker A] [Worker B] [Worker C]            [Generator] ──► [Evaluator] ───┘ (Iterate until Pass)

1. Prompt Chaining（鏈式工作流）

• 邏輯：將大任務拆解為一連串前後相依的子步驟（$A \to B \to C$）。前一個節點的輸出，經過驗證後作為下一個節點的輸入。
• 場景：文章撰寫（大綱 $\to$ 初稿 $\to$ 翻譯 $\to$ 排版）。

2. Routing（路由分流工作流）

• 邏輯：由一個輕量級的分類器模型（Router）分析 User Input，並將請求精準導向至最適合的專用 Agent 或 Prompt 處理。
• 場景：客服分流（退款請求 $\to$ 退款專用 Agent；技術故障 $\to$ 軟體工程 Agent）。

3. Parallelization（平行協同工作流）

• 邏輯：將無先後依賴關係的子任務同時發送給多個 Worker 平行處理，最後由 Aggregator 節點進行結果合併。
• 場景：大規模代碼庫審查、同時檢索 5 個不同資料源並進行交叉比對。

4. Orchestrator-Worker（主從指揮工作流）

• 邏輯：由 Orchestrator 動態分析任務、拆解子工作、建立 Worker 清單並發派執行，最後收攏成果。
• 場景：複雜專案開發、深度競品分析報告生成。

5. Evaluator-Optimizer（評估與迭代優化迴圈）

• 邏輯：包含生成者（Generator）與審查者（Evaluator）。生成結果若未通過審查者的標準，附帶批判建議（Feedback）回流重新生成，直到通過邊界條件。
• 場景：代碼生成與自動單元測試除錯、嚴謹合約審查。

三、 Python 實戰：使用 LangGraph 構建生產級 Agent Workflow

在 Python 生態系中，LangGraph 是目前建構可控 Agent Workflow 的業界標準框架。它將工作流抽象為 StateGraph（狀態圖），透過 Nodes（節點/Agent） 與 Edges（邊/條件跳轉） 來精確控制狀態演進。

以下範例實作一個包含 「需求分類 Router $\to$ 專用處理 Node $\to$ 品質審查 Evaluator 迴圈」 的完整 Workflow：

import json
from typing import TypedDict, Literal, Optional
from pydantic import BaseModel, Field
from openai import OpenAI
from langgraph.graph import StateGraph, START, END

client = OpenAI()

# ==========================================
# 1. 定義全域共享狀態 (State Schema)
# ==========================================
class WorkflowState(TypedDict):
    user_query: str                  # 原始需求
    category: Optional[str]          # Router 分類結果
    draft_response: Optional[str]    # 處理節點產出的初稿
    eval_score: Optional[int]        # 評估節點打分 (0-100)
    feedback: Optional[str]          # 評估建議
    retry_count: int                 # 重試次數防護

# ==========================================
# 2. 定義 Nodes (工作流執行節點)
# ==========================================
def router_node(state: WorkflowState) -> dict:
    """Node 1: 分類器，決定走向哪一條處理支線"""
    print("🚦 [Node: Router] 分析需求類型中...")
    query = state["user_query"]
    
    prompt = f"請將以下需求分類為 'TECHNICAL' (技術/代碼類) 或 'GENERAL' (一般諮詢類)：\n{query}"
    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}]
    )
    category = "TECHNICAL" if "TECHNICAL" in res.choices[0].message.content.upper() else "GENERAL"
    print(f"   └─ 路由判定結果: {category}")
    return {"category": category, "retry_count": 0}

def tech_agent_node(state: WorkflowState) -> dict:
    """Node 2A: 技術專屬 Agent"""
    print("💻 [Node: Tech Agent] 撰寫技術解答與程式碼...")
    prompt = f"針對技術問題：{state['user_query']}\n"
    if state.get("feedback"):
        prompt += f"修改建議：{state['feedback']}\n請重新修訂解答。"

    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一位資深架構師，請給出高質感、附帶程式碼範例的解答。"},
            {"role": "user", "content": prompt}
        ]
    )
    return {"draft_response": res.choices[0].message.content}

def general_agent_node(state: WorkflowState) -> dict:
    """Node 2B: 一般諮詢 Agent"""
    print("💬 [Node: General Agent] 撰寫一般回應...")
    res = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": state["user_query"]}]
    )
    return {"draft_response": res.choices[0].message.content}

class EvaluationResult(BaseModel):
    score: int = Field(description="品質打分 0-100")
    feedback: str = Field(description="評語與改進建議")

def evaluator_node(state: WorkflowState) -> dict:
    """Node 3: 品質審查節點 (Evaluator)"""
    print("🧐 [Node: Evaluator] 審查回答品質中...")
    prompt = (
        f"原始需求：{state['user_query']}\n"
        f"當前回答：\n{state['draft_response']}\n\n"
        f"請審查回答是否完整且技術準確。如果通過給予 85 分以上，未通過請給出具體修改建議。"
    )
    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format=EvaluationResult
    )
    res: EvaluationResult = completion.choices[0].message.parsed
    print(f"   └─ 評分: {res.score} | 建議: {res.feedback}")
    
    return {
        "eval_score": res.score,
        "feedback": res.feedback,
        "retry_count": state["retry_count"] + 1
    }

# ==========================================
# 3. 定義 Conditional Edges (條件跳轉邏輯)
# ==========================================
def route_decision(state: WorkflowState) -> Literal["tech_agent", "general_agent"]:
    """根據 Router 分類跳轉"""
    if state["category"] == "TECHNICAL":
        return "tech_agent"
    return "general_agent"

def quality_check_decision(state: WorkflowState) -> Literal["tech_agent", "__end__"]:
    """根據 Evaluator 打分與重試次數決定是否退回修正"""
    if state["eval_score"] >= 85 or state["retry_count"] >= 3:
        return END  # 通過或到達上限，結束工作流
    print("⚠️ 品質未達標，觸發退回重修迴圈 (Refinement Loop)！")
    return "tech_agent"

# ==========================================
# 4. 組裝 LangGraph Workflow 圖譜
# ==========================================
workflow = StateGraph(WorkflowState)

# 新增節點 (Nodes)
workflow.add_node("router", router_node)
workflow.add_node("tech_agent", tech_agent_node)
workflow.add_node("general_agent", general_agent_node)
workflow.add_node("evaluator", evaluator_node)

# 新增邊與邏輯 (Edges & Conditional Edges)
workflow.add_edge(START, "router")

# 分流跳轉
workflow.add_conditional_edges(
    "router",
    route_decision,
    {
        "tech_agent": "tech_agent",
        "general_agent": "general_agent"
    }
)

# 一般諮詢直接結束，技術支線進入 Evaluator
workflow.add_edge("general_agent", END)
workflow.add_edge("tech_agent", "evaluator")

# 審查迴圈跳轉
workflow.add_conditional_edges(
    "evaluator",
    quality_check_decision,
    {
        END: END,
        "tech_agent": "tech_agent"
    }
)

# 編譯工作流
app = workflow.compile()

# ==========================================
# 5. 執行 Workflow
# ==========================================
if __name__ == "__main__":
    initial_input = {
        "user_query": "請用 C++ 寫一個高效能的 Safe Singleton 類別，並解釋 double-checked locking 與 std::call_once 的差別。"
    }
    
    print("🚀 === Agent Workflow 啟動 ===")
    final_state = app.invoke(initial_input)
    
    print("\n✅ === Workflow 執行完畢 ===")
    print("📄 【最終產出結果】:\n")
    print(final_state["draft_response"])

四、 設計高可靠度 Agent Workflow 的黃金法則

1. State minimalization（狀態最小化）：

• 在 State 中只傳遞必要的數據，避免把幾萬字的中間 Log 無腦 append 到 State 中，造成下游節點 Token 暴漲。

2. Deterministic Guardrails（強依賴硬邏輯護欄）：

• 不要讓 LLM 決定關鍵的商務邏輯跳轉（例如：轉帳金額 > 10,000）。使用標準 Python 代碼條件判斷做邊界檢查，LLM 只負責內容生成與語意提取。

3. Human-in-the-Loop Interruption（人類介入點）：

• 在涉及高風險操作的 Node（如刪除資料庫、發送郵件）之前，設定 Pause 機制，等待人類授權才恢復運行。

4. Timeouts & Circuit Breakers（熔斷機制）：

• 在所有帶有 Loop（如 Reflection / Refinement）的 Workflow 中，必須加上 max_retries 硬上限，防止死迴圈消耗預算。